package com.azh.server.vm;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.azh.server.common.AbstractQuoteCache;
import com.azh.server.datam.DataManager;
import com.azh.server.tx.TxManager;
import com.azh.server.tx.TxManagerImpl;
import com.azh.server.util.ExceptionUtils;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-30
 */
public class VersionManagerImpl extends AbstractQuoteCache<Entry> implements VersionManager {

    TxManager tx;
    DataManager dm;
    Map<Long, Tx> activeTx;
    Lock lock;
    LockTable lockTable;


    public VersionManagerImpl(TxManager tx, DataManager dm) {
        super(0);
        this.tx = tx;
        this.dm = dm;
        this.activeTx = new HashMap<>();
        activeTx.put(TxManagerImpl.SUPER_TX_ID, Tx.newTx(TxManagerImpl.SUPER_TX_ID, 0, null));
        this.lock = new ReentrantLock();
        this.lockTable = new LockTable();
    }

    public void releaseEntry(Entry entry) {
        super.release(entry.getUid());
    }


    @Override
    public byte[] read(long txId, long uid) throws Exception {
        lock.lock();
        Tx t = activeTx.get(txId);
        lock.unlock();

        if (t.err != null) {
            throw t.err;
        }
        Entry entry = null;
        try {
            entry = super.get(uid);
        } catch(Exception e) {
            if(e.equals(new RuntimeException("Null entry!"))) {
                return null;
            } else {
                throw e;
            }
        }
        try {
            if(Visibility.isVisible(tx, t, entry)) {
                return entry.data();
            } else {
                return null;
            }
        } finally {
            entry.release();
        }
    }

    @Override
    public long insert(long txId, byte[] data) throws Exception {
        lock.lock();
        Tx t = activeTx.get(txId);
        lock.unlock();

        if(t.err != null) {
            throw t.err;
        }

        byte[] raw = Entry.wrapEntryRaw(txId, data);
        return dm.insert(txId, raw);
    }

    @Override
    public boolean delete(long txId, long uid) throws Exception {
        lock.lock();
        Tx t = activeTx.get(txId);
        lock.unlock();

        if(t.err != null) {
            throw t.err;
        }
        Entry entry = null;
        try {
            entry = super.get(uid);
        } catch(Exception e) {
            if (e.equals(new RuntimeException("Null entry!"))) {
                return false;
            } else {
                throw e;
            }
        }
        try {
            if(!Visibility.isVisible(tx, t, entry)) {
                return false;
            }
            Lock l = null;
            try {
                l = lockTable.add(txId, uid);
            } catch(Exception e) {
                t.err = new RuntimeException("Concurrent update issue!");
                internAbort(txId, true);
                t.autoAborted = true;
                throw t.err;
            }
            if(l != null) {
                l.lock();
                l.unlock();
            }

            if(entry.getTxMax() == txId) {
                return false;
            }

            if(Visibility.isVersionSkip(tx, t, entry)) {
                t.err = new RuntimeException("Concurrent update issue!");
                internAbort(txId, true);
                t.autoAborted = true;
                throw t.err;
            }

            entry.setTxMax(txId);
            return true;

        } finally {
            entry.release();
        }
    }

    @Override
    public long begin(int level) {
        lock.lock();
        try {
            long txId = tx.begin();
            Tx t = Tx.newTx(txId, level, activeTx);
            return txId;
        }finally {
            lock.unlock();
        }
    }

    @Override
    public void commit(long txId) throws Exception {
        lock.lock();
        Tx t = activeTx.get(txId);
        lock.unlock();

        try {
            if(t.err != null) {
                throw t.err;
            }
        } catch(NullPointerException n) {
            ExceptionUtils.getMessage(n);
        }

        lock.lock();
        activeTx.remove(txId);
        lock.unlock();

        lockTable.remove(txId);
        tx.commit(txId);
    }

    @Override
    public void abort(long txId) {
        internAbort(txId, false);
    }

    @Override
    protected Entry getForDB(long key) throws Exception {
        Entry entry = Entry.loadEntry(this, key);
        try {
            Optional<Entry> oEntry = Optional.of(entry);
        } catch (Exception e) {
            throw new RuntimeException("Null entry!");
        }
        return entry;
    }

    @Override
    protected void releaseForDB(Entry obj) {
        obj.release();
    }

    private void internAbort(long xid, boolean autoAborted) {
        lock.lock();
        Tx t = activeTx.get(xid);
        if(!autoAborted) {
            activeTx.remove(xid);
        }
        lock.unlock();

        if(t.autoAborted) return;
        lockTable.remove(xid);
        tx.abort(xid);
    }
}
